package com.coolpad.job.device.util;

import com.coolpad.Utils.GeoHashUtils;
import com.coolpad.job.device.model.GeoFullInfoPO;
import org.apache.hadoop.hbase.client.Put;

import java.util.List;
import java.util.stream.Collectors;

/**
 * 功能描述:将model转换为put<br/>
 *
 * @Author:shengtao_yang
 * @Date: 2021/1/8 11:22
 * @since: 1.0.0
 */
public class HBaseConvertUtils {

    /**
     * 将地理位置信息写入HBase
     * @param pos
     * @return
     */
    public static List<Put> toPuts(List<GeoFullInfoPO> pos){

        return pos.stream()
                .map(po -> {
                    Put put = new Put(GeoHashUtils.geoHash(po.getLon(),po.getLat()).getBytes());
                    byte[] cf_location = Constants.HBASE_GEO_FIELD.cf_location.getBytes();
                    byte[] lon = Constants.HBASE_GEO_FIELD.lon.getBytes();
                    byte[] lat = Constants.HBASE_GEO_FIELD.lat.getBytes();
                    byte[] cf_info = Constants.HBASE_GEO_FIELD.cf_info.getBytes();
                    byte[] partition_id = Constants.HBASE_GEO_FIELD.partition_id.getBytes();
                    byte[] province_id = Constants.HBASE_GEO_FIELD.province_id.getBytes();
                    byte[] province_name = Constants.HBASE_GEO_FIELD.province_name.getBytes();
                    byte[] city_id = Constants.HBASE_GEO_FIELD.city_id.getBytes();
                    byte[] city_name = Constants.HBASE_GEO_FIELD.city_name.getBytes();
                    byte[] county_id = Constants.HBASE_GEO_FIELD.county_id.getBytes();
                    byte[] county_name = Constants.HBASE_GEO_FIELD.county_name.getBytes();
                    byte[] town_id = Constants.HBASE_GEO_FIELD.town_id.getBytes();
                    byte[] town_name = Constants.HBASE_GEO_FIELD.town_name.getBytes();
                    put.addColumn(cf_location,lon,po.getLon().getBytes());
                    put.addColumn(cf_location,lat,po.getLat().getBytes());
                    put.addColumn(cf_info,partition_id,po.getPartitionId().toString().getBytes());
                    put.addColumn(cf_info,province_id,po.getProvinceId().toString().getBytes());
                    put.addColumn(cf_info,province_name,po.getProvinceName().getBytes());
                    put.addColumn(cf_info,city_id,po.getCityId().toString().getBytes());
                    put.addColumn(cf_info,city_name,po.getCityName().getBytes());
                    put.addColumn(cf_info,county_id,po.getCountyId().toString().getBytes());
                    put.addColumn(cf_info,county_name,po.getCountyName().getBytes());
                    put.addColumn(cf_info,town_id,po.getTownId().toString().getBytes());
                    put.addColumn(cf_info,town_name,po.getTownName().getBytes());
                    return put;
                })
                .collect(Collectors.toList());
    }
}
